package com.epoint.flinkdemo.transform;

import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Random;

/**
 * @author liufl
 * @version [版本号, 21-4-8]
 */
public class ReduceTest
{
    private static final Logger LOG = LoggerFactory.getLogger(KeybyTest.class);

    private static final String[] TYPE = {"苹果", "梨", "西瓜", "葡萄", "火龙果"};

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Tuple2<String, Integer>> sourceStream = env
                .addSource(new SourceFunction<Tuple2<String, Integer>>()
                {
                    private volatile boolean isRunning = true;

                    private final Random random = new Random();

                    @Override
                    public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                        while (isRunning) {
                            Thread.sleep(1000);
                            ctx.collect(new Tuple2<>(TYPE[random.nextInt(TYPE.length)], 1));
                        }
                    }

                    @Override
                    public void cancel() {
                        isRunning = false;
                    }
                }, "order-info");

        // sourceStream.keyBy(0).print();

        // keyBy之后进行分组
        sourceStream.keyBy(0).reduce(new ReduceFunction<Tuple2<String, Integer>>()
        {
            // 将上一元素与当前元素相加后，返回给下一元素处理
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2)
                    throws Exception {
                LOG.info("value1 == " +value1.toString());
                return new Tuple2<>(value1.f0, value1.f1 + value2.f1);
            }
        }).print();

        env.execute();
    }
}
